package com.ossfk.net.client;

import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.StringUtils;

import com.ossfk.net.client.adapter.AbstractClientAdapter;
import com.ossfk.net.config.NetClientConfig;
import com.ossfk.net.session.TcpSession;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoop;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.MessageToByteEncoder;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;

/**
 * 
 * @ClassName: AbstractTcpClient
 * @Description: TCP客户端
 * @author Liujiefu
 * @date 2020-03-13 17:09:13
 */
@Slf4j
public abstract class AbstractTcpClient {
	/**
	 * 会话
	 */
	private TcpSession session = null;
	/**
	 * 客户端配置
	 */
	protected NetClientConfig clientConfig = null;

	/**
	 * @Description: 默认连接本机端口8000
	 * @author Liujiefu
	 * @date 2020-03-13 16:26:19
	 */
	public AbstractTcpClient() {
		this(null);
	}

	/**
	 * @Description: 创建TCP客户端
	 * @param clientConfig
	 * @author Liujiefu
	 * @date 2020-03-13 16:28:12
	 */
	public AbstractTcpClient(NetClientConfig clientConfig) {
		super();
		if (clientConfig == null)
			clientConfig = new NetClientConfig();
		this.clientConfig = clientConfig;
		this.clientConfig.setClient(this);
	}

	private synchronized void doConnect(Bootstrap bootstrap, EventLoopGroup eventLoopGroup) {
		ChannelFuture future = null;
		try {
			init();
			if (bootstrap != null) {
				// 设置线程池
				bootstrap.group(eventLoopGroup);
				// 设置远程连接地址
				bootstrap.remoteAddress(this.clientConfig.getSocketAddress());
				// 设置socket工厂
				bootstrap.channel(NioSocketChannel.class);
				// 设置管道
				bootstrap.handler(new ChannelInitializer<Channel>() {
					@Override
					protected void initChannel(Channel ch) throws Exception {
						ch.pipeline().addLast(
								new IdleStateHandler(clientConfig.getReaderIdleTimeSeconds(), 0, 0, TimeUnit.SECONDS));
						DelimiterBasedFrameDecoder frameDecoder = getFrameDecoder();
						if (frameDecoder != null)
							ch.pipeline().addLast(frameDecoder);
						ByteToMessageDecoder decoder = getMsgDecoder();
						if (decoder != null) {
							ch.pipeline().addLast("decoder", decoder);
						}
						MessageToByteEncoder<?> encoder = getMsgEncoder();
						if (encoder != null) {
							ch.pipeline().addLast("encoder", encoder);
						}
						AbstractClientAdapter adapter = getChannelAdapter();
						if (adapter != null) {
							adapter.setConfig(clientConfig);
							ch.pipeline().addLast(adapter);
						}
					}
				});
				// 是否保持连接
				if (this.clientConfig.isKeepAlive())
					bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
				log.debug("Connecting to the server,SocketAddress={}", clientConfig.getSocketAddress().toString());
				future = bootstrap.connect();
				// 自动重连
				future.addListener((ChannelFuture futureListener) -> {
					final EventLoop eventLoop = futureListener.channel().eventLoop();
					if (!futureListener.isSuccess()) {
						log.info("{} disconnect", clientConfig.getName());
						if (clientConfig.isAutoConnect()) {
							long autoConnectSeconds = clientConfig.getAutoConnectSeconds();
							log.info("reconnect after {} seconds", autoConnectSeconds);
							eventLoop.schedule(() -> connect(), autoConnectSeconds, TimeUnit.SECONDS);
						}
					} else {
						// 连接成功，保存session
						session = createSession(futureListener.channel());
						log.info("{} connection successful,SocketAddress={}", clientConfig.getName(),
								clientConfig.getSocketAddress().toString());
					}
				});
				// 阻塞
//				future.channel().closeFuture().sync();
			}
		} catch (Exception e) {
			log.debug("{} connect error:{}", clientConfig.getName(), e.getMessage());
			log.error(e.getMessage(), e);
		}
	}

	protected void init() {
		log.info("TcpClient {} init ...{}:{}", clientConfig.getName(), clientConfig.getSocketAddress().getHostString(),
				clientConfig.getSocketAddress().getPort());
	}

	/**
	 * 创建会话
	 * 
	 * @param channel
	 * @return
	 * @author Liujiefu
	 * @date 2020-03-17 11:42:37
	 */
	protected TcpSession createSession(Channel channel) {
		session = TcpSession.buildSession(channel);
		// 标记客户端手机号码
		if (StringUtils.isNotBlank(this.clientConfig.getPhoneNum())) {
			session.setPhoneNum(this.clientConfig.getPhoneNum());
		}
		// 标记客户端设备号
		if (StringUtils.isNotBlank(this.clientConfig.getDevid())) {
			session.setPhoneNum(this.clientConfig.getDevid());
		}
		return session;
	}

	/**
	 * 连接server端
	 * 
	 * @param host server端的ip
	 * @param port server端的端口
	 * @return
	 */
	public synchronized void connect() {
		// 连接不存在或者已经断开
		this.doConnect(new Bootstrap(), new NioEventLoopGroup());
	}

	/**
	 * 客户端是否已连接
	 * 
	 * @return
	 * @author Liujiefu
	 * @date 2020-03-17 11:33:25
	 */
	public boolean isConnect() {
		if (this.session != null && this.session.getChannel() != null)
			return this.session.getChannel().isActive();
		else
			return false;
	}

	/**
	 * 获取客户端session
	 * 
	 * @return
	 * @author Liujiefu
	 * @date 2020-03-18 11:16:55
	 */
	public TcpSession getSession() {
		return this.session;
	}

	/**
	 * 清理客户端session
	 * 
	 * @param msg
	 * @author Liujiefu
	 * @date 2020-03-18 11:17:05
	 */
	public void clearSession() {
		if (this.session != null)
			this.session.setDead(true);
		this.session = null;
	}

	/**
	 * 发送消息
	 * 
	 * @param msg
	 * @author Liujiefu
	 * @date 2020-03-17 11:58:18
	 */
	public void sendMsg(Object msg) {
		if (this.session != null && !this.session.isDead())
			this.session.getChannel().writeAndFlush(msg);
	}

	/**
	 * 指定session发送消息
	 * 
	 * @param session
	 * @param msg
	 * @author Liujiefu
	 * @date 2020-03-17 11:58:01
	 */
	public void sendMsg(TcpSession session, Object msg) {
		if (session != null && !session.isDead())
			session.getChannel().writeAndFlush(msg);
	}

	/**
	 * 分包器
	 * 
	 * @return
	 * @author Liujiefu
	 * @date 2020-03-17 13:37:35
	 */
	protected abstract DelimiterBasedFrameDecoder getFrameDecoder();

	/**
	 * 解码器
	 * 
	 * @return
	 * @author Liujiefu
	 * @date 2020-03-17 13:40:20
	 */
	protected abstract ByteToMessageDecoder getMsgDecoder();

	/**
	 * 编码器
	 * 
	 * @return
	 * @author Liujiefu
	 * @date 2020-03-17 13:40:28
	 */
	protected abstract MessageToByteEncoder<?> getMsgEncoder();

	/**
	 * 通道处理适配
	 * 
	 * @return
	 * @author Liujiefu
	 * @date 2020-03-17 13:41:15
	 */
	protected abstract AbstractClientAdapter getChannelAdapter();

}
